package com.sjsu.aws.doublebarrier;

import java.lang.management.ManagementFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import com.sjsu.aws.utils.TimeUtil;
import com.sjsu.aws.utils.UUIDGenerator;

/**
 * 
 * @author Pawar Deepali
 * 
 */
public class DoubleBarrierLifeCycle {
	private static final Log	LOG	     = LogFactory
	                                             .getLog(DoubleBarrierLifeCycle.class);
	private Object	         sharedMutex	= new Object();
	private Integer	         participantCount;
	private String	         participantId;
	
	private String	         barrierId	 = null;
	private DoubleBarrier	         db	         = new DoubleBarrier();
	
	/**
	 * 
	 * @param participantCount
	 * @param participantId
	 * @param db
	 * @param timeout
	 * @throws Exception
	 */
	public DoubleBarrierLifeCycle(Integer participantCount,
	        String participantId, DoubleBarrier db, long timeout)
	        throws Exception {
		this.participantCount = participantCount;
		this.participantId = participantId;
		this.barrierId = db.getBarrierId();
		db.setUuid(UUIDGenerator.getUUID());
		db.setEndTimestamp(TimeUtil.getEndTime(timeout));
		createWhenThereIsNoNode(db);
	}
	
	public boolean ensureAllParticipantsAreReady() throws Exception {
		while (true) {
			synchronized (sharedMutex) {
				db.setBarrierId(barrierId);
				DoubleBarrierServiceImpl doubleBarrierServiceImpl = new DoubleBarrierServiceImpl();
				int noOfChildren = doubleBarrierServiceImpl
				        .getNoOfBarrierPaticipants(db);
				if (noOfChildren == participantCount) {
					LOG.info("Participant are ready");
					return true;
				} else {
					waitUntilThereIsEventNotificationFromZooKeeper();
				}
			}
		}
	}
	
	public void ensureAllParticipantsStartTheirJobs() throws Exception {
		while (true) {
			synchronized (sharedMutex) {
				db.setBarrierId(barrierId);
				//fetch number of barriers and compare with participant count
				DoubleBarrierServiceImpl doubleBarrierServiceImpl = new DoubleBarrierServiceImpl();
				int noOfChildren = doubleBarrierServiceImpl
				        .getNoOfBarrierPaticipants(db);
				if (noOfChildren == participantCount) {
					LOG.info("Starting job ");
					return;
				} else {
					waitUntilThereIsEventNotificationFromZooKeeper();
				}
			}
		}
	}
	
	public boolean leave() throws Exception {
		DoubleBarrier db = new DoubleBarrier();
		db.setBarrierId(barrierId);
		db.setParticipantId(this.participantId);
		DoubleBarrierServiceImpl doubleBarrierServiceImpl = new DoubleBarrierServiceImpl();
		doubleBarrierServiceImpl.removeParticipantFromBarrier(db);
		while (true) {
			synchronized (sharedMutex) {
				int noOfChildren = doubleBarrierServiceImpl
				        .getNoOfBarrierPaticipants(db);
				
				if (noOfChildren == 0) {
					LOG.info("Children already left");
					return true;
				} else {
					waitUntilThereIsEventNotificationFromZooKeeper();
				}
			}
		}
	}
	
	private String uniqueId() {
		return ManagementFactory.getRuntimeMXBean().getName()
		        + Thread.currentThread().getId();
	}
	
	private void createWhenThereIsNoNode(DoubleBarrier db) throws Exception {
		DoubleBarrierServiceImpl doubleBarrierServiceImpl = new DoubleBarrierServiceImpl();
		doubleBarrierServiceImpl.addParticipantToBarrier(db);
	}
	
	private void wakeUpWatingThread() {
		synchronized (sharedMutex) {
			sharedMutex.notify();
		}
	}
	
	private void waitUntilThereIsEventNotificationFromZooKeeper()
	        throws InterruptedException {
		sharedMutex.wait();
	}
}
